<?php
/**
 * crm系统推送数据到 MQ，EMS系统接受后存表
 *
 * 需要开启定时任务消费 mq 队列， 单进程消费
 *
 *
 * User: guoxiong
 * Date: 2018/4/16
 * Time: 10:05
 */

namespace app\controllers\crontab;

use app\components\mq\MqComponent;
use app\components\redis\RedisLock;
use app\components\redis\RedisMarketingEmail;
use app\models\crontab\CrmPush\UserLabel;
use app\models\crontab\CrmPush\UserLabelQueueStatus;
use yii;
use app\models\Elog;

class CrmPushController extends \app\base\Controller
{
    /**
     * redis连接
     * @var object
     */
    private $redis;
    /**
     * 初始化
     */
    public function init()
    {
        yii::$app->db->enableSlaves = false;        // 禁用从库查询，从主库查询数据。

        $this->redis = new RedisMarketingEmail();
    }

    /**
     * 消费 crm mq 里的数据，定时任务执行
     * /crontab/crm-push/user-label?module_name=marketing_email
     */
    public function actionUserLabel()
    {
        set_time_limit(0);

        $redisLock = new RedisLock($this->redis, "crm_push_user_label", 300);
        if (false === $redisLock->tryLock()) {
            exit("获得redis锁失败，退出重发进程");
        }

        try{

            $userLabelModel = new UserLabel();
            MqComponent::receiveQueueData('labelUsers_EMP',$userLabelModel,'handler');

        }catch(\Exception $exp){
            echo $exp->getMessage();
        }
    }


    /**
     * 用户标签状态计算，定时任务执行
     *
     * 1、统计同步是否完成
     *
     * /crontab/crm-push/user-label-status?module_name=marketing_email
     */
    public function actionUserLabelStatus()
    {
        set_time_limit(0);

        $redisLock = new RedisLock($this->redis, "crm_push_user_label_status", 300);
        if (false === $redisLock->tryLock()) {
            exit("获得redis锁失败，退出重发进程");
        };

        try{

            (new UserLabelQueueStatus())->updateFinishQueue();

        }catch(\Exception $exp){
            echo $exp->getMessage();
        }

    }

    /*
   *  @Explain : 获取网站的订单
   *  @Data ['sql' => '符合条件的sql语句','sqlcount' => '查看总数语句','site_name' => '网站名称','site_id' => '网站id']
   *  @Author xianxuefeng
   *  @Date 2018/8/14
   *  @Return bool
   */
    public function actionInsertUserLable()
    {

        set_time_limit(0);
        $fp = fopen(ROOT_PATH."/insertuserlabel_lock.txt", "r");
// 加锁（如果改成flock($fp, LOCK_EX | LOCK_NB)，demo2.php会直接返回2，否则会等待demo.php执行完返回1）

        $method = yii::$app->request->getMethod();
        yii::$app->request->$method("site");
        $formData = $_REQUEST;
        if (! isset($formData['json']) || empty($formData['json'])) {
            Elog::logCrontab("\n\n in the InsertUserLable method,\n params error ".\GuzzleHttp\json_encode($formData)  ,"crmpush","info");
            exit("\n\n failed.....\n");
        }
        $data = \GuzzleHttp\json_decode($formData['json'],true);
        if (empty($data['count']) || empty($data['sql']) || empty($data['site_name'])) {
            Elog::logCrontab("\n\n in the InsertUserLable method,\n data error ".\GuzzleHttp\json_encode($data)  ,"crmpush","info");
            exit("\n\n failed.....\n");
        }
        $redis = new RedisMarketingEmail();
        if( ! flock($fp, LOCK_EX | LOCK_NB))  //文件锁住中
        {   $data['retry_count'] -= 1;
            $redis->hincrby($data['key'],$data['field'],\GuzzleHttp\json_encode($data));    // 队里一个key.
            fclose($fp);  //关闭文件
            exit("\n\n busy.....\n");
        }
        flock($fp,LOCK_UN);
        sleep(0.1);
        $info = $redis->hget($data['key'],$data['field']);
        //$info = $redis->hdel($data['key'],$data['field']);
        if (empty($info)) {
            Elog::logCrontab("\n\n in the InsertUserLable method, hget ({$data['key']},{$data['field']}) failed ".\GuzzleHttp\json_encode($data)  ,"crmpush","info");
            exit("\n\n failed.....\n");
        }
        $siteName = $data['push_data']['site'];
        $pageStart = 0;

        $offset = $redis->hget($data['key']."_offset",$data['field']);
        $offset = (empty($offset)) ? 0 : $offset;
        $size = 20000; // 每次执行的大小

        while (true) {
            $tmpSql = $data['sql'] . " limit {$size} OFFSET {$offset} ";
            $userList = yii::$app->db->createCommand($tmpSql)->queryAll();
            if (empty($userList) || !is_array($userList)) {
                break;
            }
            $userArr = [];
            foreach ($userList as $val) {
                $userArr[] = $val['user_id'];
            }
            $data['push_data']['user_ids'] = $userArr;
            self::addQueueDetail($data['queue_info']['queue_detail_name'], $data['push_data']);
            // 更新位移动
            ++$pageStart;
            $offset = $pageStart * $size;
            $redis->hset($data['key']."_offset",$data['field'],$offset);
        }
        $status = 2;
        if ($siteName == 'gearbest' || $siteName == 'zaful') {   // 目前gearbest 和 zaful 数据导入要同步到es才算完成
            $status = 1;
        }
        // 更新标签添加状态
        $updateData = [
            'status' => $status,
            'update_user' => $data['update_user'],
            'update_time' => yii::$app->systime->time(),
        ];
        $updateWhere = [
            'queue_id' => $data['queue_info']['queue_id']
        ];
        $affectedRows = Yii::$app->db->createCommand()->update('crm_user_label_queue', $updateData, $updateWhere)->execute();
        if (empty($affectedRows)) {
            exit("\n\n done.....\n");
        }
        $redis->hdel($data['key'],$data['field']);
        $redis->hdel($data['key']."_offset",$data['field']);
        fclose($fp);  //关闭文件
        exit("\n\n done.....\n");
    }

    /**
     * @Explain : 获取query_id 。手动导入query_id 设置成大于100万的id下标
     * @Author xianxuefeng
     * @Date 2018/8/14
     * @Return int | null
     */
    protected function getQueryId()
    {
        yii::$app->db->enableSlaves = false;        // 禁用从库查询，从主库查询数据。防止生成子分组时主从不同步

        $sql = "select queue_id from crm_user_label_queue order by queue_id desc limit 1";
        $crmUserLabelList = Yii::$app->db->createCommand($sql)->queryOne();
        if (empty($crmUserLabelList)) {
            return null;
        }
        if ($crmUserLabelList['queue_id'] < 1000000) {
            $crmUserLabelList['queue_id'] = 1000000;  //没有一百万，赋值一百万
        }
        return $crmUserLabelList['queue_id'] + 1;
    }


    /**
     *
     */
    private function addQueue($data)
    {

        $sql = " SELECT * FROM crm_user_label_queue WHERE 1=1 AND queue_id = :queue_id ";
        $params['queue_id'] = $data['queue_id'];
        $queueInfo = Yii::$app->db->createCommand($sql, $params)->queryOne();
        if (!empty($queueInfo)) {
            $queueInfo['queue_detail_name'] = 'crm_user_label_' . $data['site'];
            return $queueInfo;
        }
        return self::_create($data);
    }

    private function _create($data)
    {
        //判断表是否存在，不存在就创建
        $siteName = $data['site'];
        $tableName = 'crm_user_label_' . $siteName;
        $res = self::createTable($tableName);
        if (empty($res)) {
            Elog::logCrontab("in the _create function ,table is empty {$tableName} ", "userGroup_{$this->time}", "info");
            return [];
        }
        $type = isset($data['type']) ? $data['type'] : 0;
        $create_user = isset($data['create_user']) ? $data['create_user'] : "system";
        $update_user = isset($data['update_user']) ? $data['update_user'] : "system";
//        $status = 2;
//        if ($siteName == 'gearbest' || $siteName == 'zaful') {   // 目前gearbest 和 zaful 数据导入要同步到es才算完成
//            $status = 1;
//        }
        $insertData = [
            'batch_id' => $data['batch_id'],
            'queue_id' => $data['queue_id'],
            'queue_name' => $data['queue_name'],
            'labels' => $data['labels'],
            'count' => $data['count'],
            'site' => $data['site'],
//            'status' => $status, //推送中
            'status' => "1", //推送中
            'queue_detail_name' => $tableName,
            'create_user' => $create_user,
            'create_time' =>  yii::$app->systime->time(),   //北京时间
            'update_user' => $update_user,
            'update_time' =>  yii::$app->systime->time(),
            'type' => $type
        ];

        $res = Yii::$app->db->createCommand()->insert('crm_user_label_queue', $insertData)->execute();
        //$lastInsertId = Yii::$app->db->getLastInsertID();

        if (empty($res)) {
            Elog::logCrontab("in the _create function ,add crm_user_label_queue info failed  " . \GuzzleHttp\json_encode($insertData), "userGroup_{$this->time}", "info");
            return [];
        }
        //创建队列详情表
        return $insertData;
    }

    /**
     * yumao
     * 2018-5-30
     * @param type $queueInfo
     */
    private function getSiteCrmDetailTableName($queueInfo)
    {
        $detailTableName = 'crm_user_label';
        $detailTableName .= '_' . $queueInfo['site'];
        return $detailTableName;
    }

    /**
     * 创建队列详情数据
     *
     * @param $tableName
     * @param $pushData
     * @throws \yii\db\Exception
     */
    private function addQueueDetail($tableName, $pushData)
    {
        if (empty($pushData['user_ids'])) {
            return false;
        }
        if (!is_array($pushData['user_ids'])) {
            return false;
        }
        $columns = [
            'queue_id',
            'batch_id',
            'site_user_id',
            'appsflyer_id',
            'create_time'
        ];
        $rows = [];
        foreach ($pushData['user_ids'] as $key => $val) {
            $detailArray = explode("|", $val);
            $detailArray[0] = intval(isset($detailArray[0]) ? $detailArray[0] : 0);
            $detailArray[1] = intval(isset($detailArray[1]) ? $detailArray[1] : 0);
            $rows[] = [
                $pushData['queue_id'],
                $pushData['batch_id'],
                $detailArray[0],
                $detailArray[1],
                yii::$app->systime->time(),
            ];
        }
        $res = Yii::$app->db->createCommand()->batchInsert($tableName, $columns, $rows)->execute();
        if (empty($res)) {
            return false;
        }
        return true;
    }

    /**
     * 创建一个新的队列详情表
     *
     * @param $tableName
     * @return bool
     * @throws \yii\db\Exception
     */
    public function createTable($tableName)
    {
        if (empty($tableName)) {
            return false;
        }
        $exist = self::_exitsTable($tableName);
        if (!empty($exist)) {
            return true;
        }

        $tableSql = "
            CREATE TABLE `{$tableName}` (
              `id` INT(11) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键',
              `queue_id` INT(11) NOT NULL DEFAULT 0 COMMENT 'crm系统队列id',
              `batch_id` INT(11) NOT NULL DEFAULT 0 COMMENT 'crm系统队列id',
              `site_user_id` BIGINT(19) NOT NULL DEFAULT 0 COMMENT '网站用户id',
              `appsflyer_id` INT(11) NOT NULL DEFAULT 0 COMMENT '设备id',
              `create_time` INT(11) NOT NULL DEFAULT 0 COMMENT '修改时间',
              PRIMARY KEY (`id`),
              KEY `idx_user` (`site_user_id`),
              KEY `idx_queue` (`queue_id`)
            ) ENGINE=INNODB  DEFAULT CHARSET=utf8 COMMENT='用户标签详情'
        ";
        Yii::$app->db->createCommand($tableSql)->execute();

        return true;
    }

    /**
     * 判断表是否存在
     *
     * @param $tableName
     * @return array|false
     */
    protected function _exitsTable($tableName)
    {
        $tableExitsSql = " SELECT table_name FROM information_schema.TABLES WHERE table_name = :table_name";
        $tableExitsParams['table_name'] = $tableName;

        return Yii::$app->db->createCommand($tableExitsSql, $tableExitsParams)->queryOne();
    }

    /**
     * 返回二维数组，指定下标，值非空的$delimiter分隔字符串,$delimiter 默认为逗号
     * 返回逗号分隔字符串
     *
     */
    private function arr2str($array, $key, $type = 'int', $delimiter = ',')
    {
        if (empty($array) || empty($key)) {
            return false;
        }
        $string = '';
        if ($type == 'int') {
            foreach ($array as $val) {
                if (isset($val[$key]) && !empty($val[$key])) {
                    $string .= $val[$key] . $delimiter;
                }
            }
        } else {
            foreach ($array as $val) {
                if (isset($val[$key]) && !empty($val[$key])) {
                    $string .= "'{$val[$key]}'$delimiter";
                }
            }
        }
        return rtrim($string, $delimiter);
    }

}
